package internal

import (
	"fmt"
	client "github.com/influxdata/influxdb1-client/v2"
	"github.com/mitchellh/mapstructure"
	"github.com/thoas/go-funk"
	"strings"
	"sync"
	"time"
	"zt-compute/config"
)

type TagValueHistory struct {
	Quality int32   `json:"quality"`
	Time    int64   `json:"time"`
	Value   float64 `json:"value"`
	Code    string  `json:"code"`
}

type TagValueHistorySlice []TagValueHistory

func (s TagValueHistorySlice) Len() int           { return len(s) }
func (s TagValueHistorySlice) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
func (s TagValueHistorySlice) Less(i, j int) bool { return s[i].Code < s[j].Code }

type HistoryQuery struct {
	OnMoment   bool     `form:"onMoment"`
	BeginTime  int64    `form:"beginTime"`
	EndTime    int64    `form:"endTime"`
	JudgeValue float64  `form:"judgeValue"`
	Type       string   `form:"type"`
	Interval   string   `form:"interval"`
	Symbol     string   `form:"symbol"`
	Tags       []string `form:"tags"`
}

func GeHistoryData(query HistoryQuery) []TagValueHistory {
	n := time.Now().UnixNano()
	rangeStr := ""
	queryTags := funk.Chunk(query.Tags, 10)
	queryStrs := make([]string, len(queryTags.([][]string)))
	fullSql := `SELECT * FROM tag_value WHERE time>=%dms AND time<=%dms %s %s GROUP BY code`
	groupSql := `SELECT %s(value) as value FROM tag_value WHERE time >=%dms AND time<%dms %s %s`
	groupSql += `GROUP BY time(%s,-8h), code fill(none)`
	for i, tags := range queryTags.([][]string) {
		tagStr := getCodeQueryStr(tags)
		switch query.Type {
		case "full":
			queryStrs[i] = fmt.Sprintf(fullSql, query.BeginTime, query.EndTime, tagStr, rangeStr)
		case "MEAN":
			groupSql = `SELECT %s(value) as "value" FROM "tag_value" WHERE time >=%dms AND time<%dms %s %s GROUP BY time(%s,-8h),code fill(linear)`
			queryStrs[i] = fmt.Sprintf(groupSql, query.Type, query.BeginTime,
				query.EndTime, tagStr, rangeStr, query.Interval)
		case "period":
			groupSql = `SELECT MAX(value) - MIN(value) as value FROM tag_value WHERE time >=%dms AND time<=%dms %s %s`
			groupSql += `GROUP BY code fill(linear)`
			queryStrs[i] = fmt.Sprintf(groupSql, query.BeginTime, query.EndTime, tagStr, rangeStr)
		default:
			queryStrs[i] = fmt.Sprintf(groupSql, query.Type, query.BeginTime,
				query.EndTime, tagStr, rangeStr, query.Interval)
		}
	}

	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr: config.MyConfig.FastDBAddress,
	})
	if err != nil {
		MyLog.Error("Error creating InfluxDB Client: ", err.Error())
	}
	defer c.Close()
	result := queryData(queryStrs, config.MyConfig.FastDBName, c)
	historyData := convertToTagValueHistory(result)
	MyLog.Debug("cost:", (time.Now().UnixNano()-n)/1e6)
	return historyData
}

func getCodeQueryStr(codes []string) string {
	return "AND (code='" + strings.Join(codes, "' OR code='") + "')"
}

func queryData(sqlQueryList []string, database string, c client.Client) []map[string]interface{} {
	var wg sync.WaitGroup
	result := make([]map[string]interface{}, 0)
	queue := make(chan []map[string]interface{}, 1)
	wg.Add(len(sqlQueryList))
	for i := 0; i < len(sqlQueryList); i++ {
		go func(sql string, database string, c client.Client) {
			MyLog.Debug(sql)
			q := client.NewQuery(sql, database, "ms")
			response, err := c.Query(q)
			var m []map[string]interface{}
			if err == nil && response.Error() == nil {
				m = GroupBy(response.Results[0])
				queue <- m
			} else {
				if response != nil {
					MyLog.Error(response.Error())
				}
			}
		}(sqlQueryList[i], database, c)
	}
	go func() {
		for rs := range queue {
			result = append(result, rs...)
			wg.Done()
		}
	}()
	wg.Wait()
	close(queue)
	return result
}

func GroupBy(result client.Result) []map[string]interface{} {
	rows := make([]map[string]interface{}, 0)
	for _, ser := range result.Series {
		for i := range ser.Values {
			m := make(map[string]interface{})
			if ser.Tags != nil {
				for k, v := range ser.Tags {
					m[k] = v
				}
			}
			for index := range ser.Columns {
				m[ser.Columns[index]] = ser.Values[i][index]
			}
			rows = append(rows, m)
		}
	}
	return rows
}

func convertToTagValueHistory(inputs []map[string]interface{}) []TagValueHistory {
	//b, _ := jsoniter.Marshal(&inputs)

	result := make(TagValueHistorySlice, len(inputs))
	err := mapstructure.Decode(inputs, &result)

	//err := jsoniter.Unmarshal(b, &result)
	if err != nil {
		MyLog.Error(err)
	}
	return result
}
